/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.broker.transaction.queue;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener;
import org.apache.rocketmq.broker.transaction.OperationResult;
import org.apache.rocketmq.broker.transaction.TransactionalMessageService;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.config.BrokerRole;

public class TransactionalMessageServiceImpl implements TransactionalMessageService {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.TRANSACTION_LOGGER_NAME);

    private TransactionalMessageBridge transactionalMessageBridge;

    private static final int PULL_MSG_RETRY_NUMBER = 1;

    private static final int MAX_PROCESS_TIME_LIMIT = 60000;
    private static final int MAX_RETRY_TIMES_FOR_ESCAPE = 10;

    private static final int MAX_RETRY_COUNT_WHEN_HALF_NULL = 1;

    private static final int OP_MSG_PULL_NUMS = 32;

    private static final int SLEEP_WHILE_NO_OP = 1000;

    /**
     * key：half消息的queueId，value是OP消息队列上下文
     */
    private final ConcurrentHashMap<Integer, MessageQueueOpContext> deleteContext = new ConcurrentHashMap<>();

    private ServiceThread transactionalOpBatchService;

    private ConcurrentHashMap<MessageQueue, MessageQueue> opQueueMap = new ConcurrentHashMap<>();

    public TransactionalMessageServiceImpl(TransactionalMessageBridge transactionBridge) {
        this.transactionalMessageBridge = transactionBridge;
        transactionalOpBatchService = new TransactionalOpBatchService(transactionalMessageBridge.getBrokerController(), this);
        transactionalOpBatchService.start();
    }


    @Override
    public CompletableFuture<PutMessageResult> asyncPrepareMessage(MessageExtBrokerInner messageInner) {
        return transactionalMessageBridge.asyncPutHalfMessage(messageInner);
    }

    @Override
    public PutMessageResult prepareMessage(MessageExtBrokerInner messageInner) {
        return transactionalMessageBridge.putHalfMessage(messageInner);
    }

    /**
     * 消息需要丢弃
     * @param msgExt
     * @param transactionCheckMax
     * @return
     */
    private boolean needDiscard(MessageExt msgExt, int transactionCheckMax) {
        // 或许消息的"TRANSACTION_CHECK_TIMES"属性值，表示消息回查了多少次
        String checkTimes = msgExt.getProperty(MessageConst.PROPERTY_TRANSACTION_CHECK_TIMES);
        int checkTime = 1;
        if (null != checkTimes) {
            checkTime = getInt(checkTimes);
            // 如果回查次数大于等于事务消息回查次数最大值：15次，则HALF消息需要丢弃
            if (checkTime >= transactionCheckMax) {
                return true;
            } else {
                // 否则回查次数+1
                checkTime++;
            }
        }
        // 将新的回查次数设置给HALF消息的"TRANSACTION_CHECK_TIMES"属性
        msgExt.putUserProperty(MessageConst.PROPERTY_TRANSACTION_CHECK_TIMES, String.valueOf(checkTime));
        // 此时HALF消息不需要丢弃
        return false;
    }

    /**
     * HALF消息存在的时间超过了日志文件保留的最大时长（默认72小时），消息需要删除
     * @param msgExt
     * @return
     */
    private boolean needSkip(MessageExt msgExt) {
        long valueOfCurrentMinusBorn = System.currentTimeMillis() - msgExt.getBornTimestamp();
        // 如果消息发送到现在的时间间隔大于日志文件保留的最大时长（默认72小时），则需要删除该消息。
        if (valueOfCurrentMinusBorn
            > transactionalMessageBridge.getBrokerController().getMessageStoreConfig().getFileReservedTime()
            * 3600L * 1000) {
            log.info("Half message exceed file reserved time ,so skip it.messageId {},bornTime {}",
                msgExt.getMsgId(), msgExt.getBornTimestamp());
            return true;
        }
        return false;
    }

    /**
     * 将HALF消息放回消息队列中，如果保存成功，则设置msgExt的queueOffset，commitlogoffset，msgId，返回true，
     * 否则返回false
     * @param msgExt
     * @param offset
     * @return
     */
    private boolean putBackHalfMsgQueue(MessageExt msgExt, long offset) {
        // 将HALF消息重新写到HALF主题中
        PutMessageResult putMessageResult = putBackToHalfQueueReturnResult(msgExt);
        if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
            // 消息队列偏移量
            msgExt.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());
            // 提交日志偏移量
            msgExt.setCommitLogOffset(putMessageResult.getAppendMessageResult().getWroteOffset());
            // msgId
            msgExt.setMsgId(putMessageResult.getAppendMessageResult().getMsgId());


            log.debug(
                "Send check message, the offset={} restored in queueOffset={} "
                    + "commitLogOffset={} "
                    + "newMsgId={} realMsgId={} topic={}",
                offset, msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getMsgId(),
                msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX),
                msgExt.getTopic());


            // true表示成功
            return true;
        } else {
            log.error(
                "PutBackToHalfQueueReturnResult write failed, topic: {}, queueId: {}, "
                    + "msgId: {}",
                msgExt.getTopic(), msgExt.getQueueId(), msgExt.getMsgId());


            // 表示失败
            return false;
        }
    }

    /**
     * 消息回查
     * @param transactionTimeout 事务消息状态未知多长时间之后开始回查消息。默认6秒
     * @param transactionCheckMax 事务消息最多回查多少次，超过此次数，则丢弃事务消息。默认15次
     * @param listener 需要回查事务消息或需要丢弃事务消息的时候，调用该类对象相应的方法，事务消息回查监听器
     */
    @Override
    public void check(long transactionTimeout, int transactionCheckMax, AbstractTransactionalMessageCheckListener listener) {
        try {
            // 事务HALF消息所在的主题名称
            String topic = TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC;
            // 获取HALF消息所属主题的消息队列列表
            Set<MessageQueue> msgQueues = transactionalMessageBridge.fetchMessageQueues(topic);
            if (msgQueues == null || msgQueues.size() == 0) {
                log.warn("The queue of topic is empty :" + topic);
                return;
            }
            log.debug("Check topic={}, queues={}", topic, msgQueues);

            // 遍历HALF消息队列中的HALF消息
            for (MessageQueue messageQueue : msgQueues) {
                // 获取当前时间
                long startTime = System.currentTimeMillis();

                // 根据HALF的MQ，获取对应的OP MQ
                MessageQueue opQueue = getOpQueue(messageQueue);

                // 获取HALF消息的偏移量
                long halfOffset = transactionalMessageBridge.fetchConsumeOffset(messageQueue);
                // 获取OP消息的偏移量
                long opOffset = transactionalMessageBridge.fetchConsumeOffset(opQueue);
                log.info("Before check, the queue={} msgOffset={} opOffset={}", messageQueue, halfOffset, opOffset);

                // 如果都没有找到，则跳过该HALF消息队列
                if (halfOffset < 0 || opOffset < 0) {
                    log.error("MessageQueue: {} illegal offset read: {}, op offset: {},skip this queue", messageQueue,
                        halfOffset, opOffset);
                    continue;
                }

                // 处理完成的OP消息偏移量集合
                List<Long> doneOpOffset = new ArrayList<>();
                // key是HALF偏移量，value是OP偏移量
                HashMap<Long, Long> removeMap = new HashMap<>();
                // key：OP消息偏移量，value：HALF消息偏移量集合
                HashMap<Long, HashSet<Long>> opMsgMap = new HashMap<Long, HashSet<Long>>();

                // 读取OP消息，解析OP消息，填充removeMap，返回读取OP消息的结果
                PullResult pullResult = fillOpRemoveMap(removeMap, opQueue, opOffset, halfOffset, opMsgMap, doneOpOffset);

                // 如果没有结果，则跳过当前HALF MQ，处理下一个HALF MQ
                if (null == pullResult) {
                    log.error("The queue={} check msgOffset={} with opOffset={} failed, pullResult is null",
                        messageQueue, halfOffset, opOffset);
                    continue;
                }


                // 单线程执行下列代码

                // 表示while循环中从当前HALF消息队列中获取消息为null的次数。
                int getMessageNullCount = 1;
                // 用于记录日志和矫正HALF MQ的偏移量
                long newOffset = halfOffset;
                // HALF消息的偏移量
                long i = halfOffset;
                // 下一个OP消息的偏移量
                long nextOpOffset = pullResult.getNextBeginOffset();
                // 用于记录日志
                int putInQueueCount = 0;
                // 暂不考虑
                int escapeFailCnt = 0;

                while (true) {
                    // 如果当前时间减去开始时间超过60秒，则退出，不再处理
                    if (System.currentTimeMillis() - startTime > MAX_PROCESS_TIME_LIMIT) {
                        log.info("Queue={} process time reach max={}", messageQueue, MAX_PROCESS_TIME_LIMIT);
                        break;
                    }
                    // 如果HALF消息的偏移量在removeMap中，则无需进行消息回查
                    if (removeMap.containsKey(i)) {
                        log.debug("Half offset {} has been committed/rolled back", i);

                        // key是HALF偏移量，value是OP偏移量
                        // 从removeMap中将HALF消息偏移量移除，返回对应的OP消息偏移量
                        Long removedOpOffset = removeMap.remove(i);

                        // key：OP消息偏移量，value：HALF消息偏移量集合
                        // 从opMsgMap中根据OP偏移量获取对应的HALF偏移量set集合，并从set集合中移除对应的HALF偏移量
                        opMsgMap.get(removedOpOffset).remove(i);

                        // 如果此时OP消息偏移量对应的set集合元素个数为0，则从opMsgMap中移除该OP偏移量对应的条目；
                        // 将OP偏移量添加到doneOpOffset集合中。
                        if (opMsgMap.get(removedOpOffset).size() == 0) {

                            // key是OP偏移量，value是HALF偏移量集合
                            // 从opMsgMap中移除当前OP偏移量对应的HALF集合
                            opMsgMap.remove(removedOpOffset);
                            // 将当前OP偏移量添加到 已完成的OP偏移量集合
                            doneOpOffset.add(removedOpOffset);
                        }
                    } else {
                        // 如果removeMap中不包含当前HALF偏移量，则需要对HALF消息进行处理
                        // 从HALF消息队列中获取偏移量对应的HALF消息
                        GetResult getResult = getHalfMsg(messageQueue, i);

                        // 获取HALF消息
                        MessageExt msgExt = getResult.getMsg();

                        // 没有从当前HALF的MQ中取到消息
                        if (msgExt == null) {
                            // 如果没有获取到消息，且获取消息为null的次数大于阈值1，则不再处理当前HALF MQ，开始处理下一个HALF MQ。
                            if (getMessageNullCount++ > MAX_RETRY_COUNT_WHEN_HALF_NULL) {
                                break;
                            }

                            // 如果拉取消息的结果状态是NO_NEW_MSG，则不再处理当前HALF MQ，开始处理下一个HALF MQ。
                            // （当前HALF消息队列没有消息了，遍历下一个HALF消息队列）
                            if (getResult.getPullResult().getPullStatus() == PullStatus.NO_NEW_MSG) {
                                log.debug("No new msg, the miss offset={} in={}, continue check={}, pull result={}", i,
                                    messageQueue, getMessageNullCount, getResult.getPullResult());
                                break;
                            } else {
                                // 偏移量错误，将HALF消息的偏移量下推一个，继续处理当前HALF消息队列中的下一条HALF消息
                                log.info("Illegal offset, the miss offset={} in={}, continue check={}, pull result={}",
                                    i, messageQueue, getMessageNullCount, getResult.getPullResult());

                                // 获取当前HALF消息队列中下一条消息的偏移量
                                i = getResult.getPullResult().getNextBeginOffset();
                                // 更新newOffset为当前HALF消息队列中下一条消息的偏移量
                                newOffset = i;
                                // 继续while循环
                                continue;
                            }
                        }

                        // 从当前HALF的MQ中取到了消息
                        // <<默认配置跳过>>
                        // 如果允许slave扮演master，并且当前broker的ID是相同brokerName的组中的最小ID，并且当前broker是slave
                        if (this.transactionalMessageBridge.getBrokerController().getBrokerConfig().isEnableSlaveActingMaster()
                            && this.transactionalMessageBridge.getBrokerController().getMinBrokerIdInGroup()
                            == this.transactionalMessageBridge.getBrokerController().getBrokerIdentity().getBrokerId()
                            && BrokerRole.SLAVE.equals(this.transactionalMessageBridge.getBrokerController().getMessageStoreConfig().getBrokerRole())
                        ) {
                            final MessageExtBrokerInner msgInner = this.transactionalMessageBridge.renewHalfMessageInner(msgExt);
                            final boolean isSuccess = this.transactionalMessageBridge.escapeMessage(msgInner);

                            if (isSuccess) {
                                escapeFailCnt = 0;
                                newOffset = i + 1;
                                i++;
                            } else {
                                log.warn("Escaping transactional message failed {} times! msgId(offsetId)={}, UNIQ_KEY(transactionId)={}",
                                    escapeFailCnt + 1,
                                    msgExt.getMsgId(),
                                    msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
                                if (escapeFailCnt < MAX_RETRY_TIMES_FOR_ESCAPE) {
                                    escapeFailCnt++;
                                    Thread.sleep(100L * (2 ^ escapeFailCnt));
                                } else {
                                    escapeFailCnt = 0;
                                    newOffset = i + 1;
                                    i++;
                                }
                            }
                            continue;
                        }

                        // 判断是否需要丢弃当前HALF消息
                        // 1. 是否达到了最大回查次数（默认15次），如果是，则丢弃，否则保留，再次尝试回查；
                        // 2. HALF消息存在的时间超过了日志文件保留的最大时长（默认72小时），消息需要删除。
                        if (needDiscard(msgExt, transactionCheckMax) || needSkip(msgExt)) {
                            // 为了避免对HALF消息进行无限回查，将消息保存到TRANS_CHECK_MAXTIME_TOPIC主题中。
                            listener.resolveDiscardMsg(msgExt);
                            // 递增HALF的MQ消息偏移量
                            newOffset = i + 1;
                            // i 递增
                            i++;
                            // 继续处理当前HALF的MQ中的下一条HALF消息
                            continue;
                        }

                        // 如果消息是当前方法开始执行后进来的消息，则不处理，跳出循环。
                        // 因为该方法开始执行后所有的HALF消息都已经处理完了。
                        if (msgExt.getStoreTimestamp() >= startTime) {
                            log.debug("Fresh stored. the miss offset={}, check it later, store={}", i,
                                new Date(msgExt.getStoreTimestamp()));
                            // 跳出当前while循环，处理下一个HALF MQ中的HALF消息
                            break;
                        }

                        // 当前时间与HALF消息发送时间的差值
                        long valueOfCurrentMinusBorn = System.currentTimeMillis() - msgExt.getBornTimestamp();
                        // 将事务超时时间设置为免疫时间，即无需事务消息回查介入的最长时间
                        long checkImmunityTime = transactionTimeout;

                        // 从消息获取"CHECK_IMMUNITY_TIME_IN_SECONDS"属性值：免疫时间，即无需事务消息回查介入的最长时间
                        String checkImmunityTimeStr = msgExt.getUserProperty(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS);

                        // 如果消息存在"CHECK_IMMUNITY_TIME_IN_SECONDS"值，则
                        if (null != checkImmunityTimeStr) {
                            // 如果checkImmunityTimeStr是-1，则返回事务超时时间，否则返回checkImmunityTimeStr的long型值
                            checkImmunityTime = getImmunityTime(checkImmunityTimeStr, transactionTimeout);

                            // 如果HALF消息等待处理的时间小于事务超时时间，表示还不需要事务消息回查的介入，按正常情况处理即可。
                            if (valueOfCurrentMinusBorn < checkImmunityTime) {
                                if (checkPrepareQueueOffset(removeMap, doneOpOffset, msgExt, checkImmunityTimeStr)) {
                                    // HALF MQ偏移量+1
                                    newOffset = i + 1;
                                    // i+1
                                    i++;
                                    // 处理当前 HALF MQ中的下一条HALF消息，因为这条HALF消息单独设置了免疫时间
                                    continue;
                                }
                            }
                        } else {
                            // 如果还在免疫时间之内，无需事务消息回查介入。
                            if (0 <= valueOfCurrentMinusBorn && valueOfCurrentMinusBorn < checkImmunityTime) {
                                log.debug("New arrived, the miss offset={}, check it later checkImmunity={}, born={}", i,
                                    checkImmunityTime, new Date(msgExt.getBornTimestamp()));
                                // 跳过当前HALF MQ，处理下一个HALF MQ中的HALF消息
                                // 后续消息肯定也在免疫时间之内，因此，跳过当前HALF MQ，处理下一个HALF MQ的消息
                                break;
                            }
                        }

                        // 获取OP消息集合
                        List<MessageExt> opMsg = pullResult == null ? null : pullResult.getMsgFoundList();

                        // 1. 如果opMsg是null，并且HALF消息已经过了免疫时间，则需要回查
                        // 2. 或者有op消息，并且OP消息集合中最后一条消息的发送时间与开始时间的差值大于事务超时时间，则需要回查
                        // 3. 或者消息时间有误，则需要回查
                        boolean isNeedCheck = opMsg == null && valueOfCurrentMinusBorn > checkImmunityTime
                            || opMsg != null && opMsg.get(opMsg.size() - 1).getBornTimestamp() - startTime > transactionTimeout
                            || valueOfCurrentMinusBorn <= -1;

                        // 如果消息需要回查
                        if (isNeedCheck) {
                            // 将HALF消息放回消息队列中，如果保存成功，则设置msgExt的queueOffset，commitlogoffset，msgId，返回true，否则返回false
                            // 为什么需要放回去呢？因为获取出来后，偏移量等都移动到下一条消息了，下次无法再次取出来。
                            // 因此再次放进去，如果这次回查失败，方便下次再次处理。
                            if (!putBackHalfMsgQueue(msgExt, i)) {
                                // 跳过当前消息
                                continue;
                            }

                            // 如果放入成功，则将放回HALF主题的消息个数+1
                            putInQueueCount++;
                            log.info("Check transaction. real_topic={},uniqKey={},offset={},commitLogOffset={}",
                                    msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC),
                                    msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX),
                                    msgExt.getQueueOffset(), msgExt.getCommitLogOffset());

                            // 使用线程池发起消息回查请求的任务
                            listener.resolveHalfMsg(msgExt);
                        } else {
                            // 如果不需要回查，则
                            // 计算下一个OP消息的偏移量
                            nextOpOffset = pullResult != null ? pullResult.getNextBeginOffset() : nextOpOffset;

                            // 读取OP消息，解析OP消息，填充removeMap，返回读取OP消息的结果
                            pullResult = fillOpRemoveMap(removeMap, opQueue, nextOpOffset,
                                    halfOffset, opMsgMap, doneOpOffset);

                            // 如果填充removeMap失败，或者没有更多OP消息，或者OP消息偏移量错误，或者没有匹配的OP消息
                            if (pullResult == null || pullResult.getPullStatus() == PullStatus.NO_NEW_MSG
                                    || pullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL
                                    || pullResult.getPullStatus() == PullStatus.NO_MATCHED_MSG) {

                                // 等待1秒钟，等待OP消息就绪
                                try {
                                    Thread.sleep(SLEEP_WHILE_NO_OP);
                                } catch (Throwable ignored) {
                                }

                            } else {
                                log.info("The miss message offset:{}, pullOffsetOfOp:{}, miniOffset:{} get more opMsg.", i, nextOpOffset, halfOffset);
                            }

                            continue;
                        }
                    }
                    // 偏移量+1，处理该HALF MQ的下一条消息
                    newOffset = i + 1;
                    // i+1
                    i++;
                }

                // 当前HALF MQ处理结束，对结果进行处理。
                // 如果newOffset与halfOffset不一致，则更新MQ到新的newOffset
                if (newOffset != halfOffset) {
                    transactionalMessageBridge.updateConsumeOffset(messageQueue, newOffset);
                }

                // 根据opOffset和已经处理好的OpOffset计算OP消息新的偏移量
                long newOpOffset = calculateOpOffset(doneOpOffset, opOffset);

                // 如果opOffset与newOpOffset不一致，则使用newOpOffset更新
                if (newOpOffset != opOffset) {
                    transactionalMessageBridge.updateConsumeOffset(opQueue, newOpOffset);
                }

                // 获取当前HALF MQ中获取指定消息
                GetResult getResult = getHalfMsg(messageQueue, newOffset);

                // 根据HALF消息从OP MQ中获取下一条OP消息
                pullResult = pullOpMsg(opQueue, newOpOffset, 1);


                // 计算当前HALF MQ的最大偏移量
                long maxMsgOffset = getResult.getPullResult() == null ? newOffset : getResult.getPullResult().getMaxOffset();
                // 计算当前OP MQ的最大偏移量
                long maxOpOffset = pullResult == null ? newOpOffset : pullResult.getMaxOffset();
                // 计算HALF的最大时间
                long msgTime = getResult.getMsg() == null ? System.currentTimeMillis() : getResult.getMsg().getStoreTimestamp();
                // 记录当前HALF MQ与OP MQ相关数据的日志
                log.info("After check, {} opOffset={} opOffsetDiff={} msgOffset={} msgOffsetDiff={} msgTime={} msgTimeDelayInMs={} putInQueueCount={}",
                        messageQueue, newOpOffset, maxOpOffset - newOpOffset, newOffset, maxMsgOffset - newOffset, new Date(msgTime),
                        System.currentTimeMillis() - msgTime, putInQueueCount);
            }
        } catch (Throwable e) {
            log.error("Check error", e);
        }
    }

    /**
     * 如果checkImmunityTimeStr是-1，则返回事务超时时间，否则返回checkImmunityTimeStr的long型值
     * @param checkImmunityTimeStr
     * @param transactionTimeout
     * @return
     */
    private long getImmunityTime(String checkImmunityTimeStr, long transactionTimeout) {
        long checkImmunityTime;

        checkImmunityTime = getLong(checkImmunityTimeStr);
        if (-1 == checkImmunityTime) {
            checkImmunityTime = transactionTimeout;
        } else {
            checkImmunityTime *= 1000;
        }
        return checkImmunityTime;
    }

    /**
     * 读取OP消息，解析OP消息，填充removeMap
     *
     * @param removeMap 需要移除的HALF消息map集合，key:halfOffset, value: opOffset.
     * @param opQueue OP消息队列
     * @param pullOffsetOfOp OP消息队列的起始偏移量
     * @param miniOffset 当前HALF消息队列的最小偏移量
     * @param opMsgMap OP消息中的HALF消息偏移量
     * @param doneOpOffset 存储处理完的OP消息偏移量
     * @return OP消息结果
     */
    private PullResult fillOpRemoveMap(HashMap<Long, Long> removeMap, MessageQueue opQueue,
                                       long pullOffsetOfOp, long miniOffset, Map<Long, HashSet<Long>> opMsgMap, List<Long> doneOpOffset) {

        // OP消息队列，OP消息队列偏移量，32
        PullResult pullResult = pullOpMsg(opQueue, pullOffsetOfOp, OP_MSG_PULL_NUMS);
        if (null == pullResult) {
            return null;
        }
        if (pullResult.getPullStatus() == PullStatus.OFFSET_ILLEGAL
            || pullResult.getPullStatus() == PullStatus.NO_MATCHED_MSG) {
            log.warn("The miss op offset={} in queue={} is illegal, pullResult={}", pullOffsetOfOp, opQueue,
                pullResult);
            transactionalMessageBridge.updateConsumeOffset(opQueue, pullResult.getNextBeginOffset());
            return pullResult;
        } else if (pullResult.getPullStatus() == PullStatus.NO_NEW_MSG) {
            log.warn("The miss op offset={} in queue={} is NO_NEW_MSG, pullResult={}", pullOffsetOfOp, opQueue,
                pullResult);
            return pullResult;
        }

        // 获取OP消息集合
        List<MessageExt> opMsg = pullResult.getMsgFoundList();
        if (opMsg == null) {
            log.warn("The miss op offset={} in queue={} is empty, pullResult={}", pullOffsetOfOp, opQueue, pullResult);
            return pullResult;
        }

        // 遍历读取到的OP消息集合
        for (MessageExt opMessageExt : opMsg) {
            if (opMessageExt.getBody() == null) {
                log.error("op message body is null. queueId={}, offset={}", opMessageExt.getQueueId(),
                        opMessageExt.getQueueOffset());
                doneOpOffset.add(opMessageExt.getQueueOffset());
                continue;
            }

            HashSet<Long> set = new HashSet<Long>();
            // 消息体字符串（OP中存储的是HALF消息所在主题消息队列的偏移量）
            String queueOffsetBody = new String(opMessageExt.getBody(), TransactionalMessageUtil.CHARSET);

            log.debug("Topic: {} tags: {}, OpOffset: {}, HalfOffset: {}", opMessageExt.getTopic(),
                    opMessageExt.getTags(), opMessageExt.getQueueOffset(), queueOffsetBody);

            // 如果消息的标签是"d"，表示是移除标签
            if (TransactionalMessageUtil.REMOVE_TAG.equals(opMessageExt.getTags())) {
                // 对消息体使用","进行切割
                String[] offsetArray = queueOffsetBody.split(TransactionalMessageUtil.OFFSET_SEPARATOR);

                for (String offset : offsetArray) {
                    // 转换为Long型偏移量值
                    Long offsetValue = getLong(offset);
                    // 如果偏移量小于HALF消息队列中的最小偏移量，则跳过
                    if (offsetValue < miniOffset) {
                        continue;
                    }
                    // 将HALF消息偏移量->OP消息偏移量 键值对放到removeMap中
                    removeMap.put(offsetValue, opMessageExt.getQueueOffset());
                    // 将HALF消息偏移量放到set集合中
                    set.add(offsetValue);
                }
            } else {
                log.error("Found a illegal tag in opMessageExt= {} ", opMessageExt);
            }

            // 如果set集合中有元素
            if (set.size() > 0) {
                // 将OP消息偏移量->set集合 键值对放到opMsgMap中
                opMsgMap.put(opMessageExt.getQueueOffset(), set);
            } else {
                // 否则将OP消息偏移量放到完成的OP消息偏移量中
                doneOpOffset.add(opMessageExt.getQueueOffset());
            }
        }

        log.debug("Remove map: {}", removeMap);
        log.debug("Done op list: {}", doneOpOffset);
        log.debug("opMsg map: {}", opMsgMap);
        return pullResult;
    }

    /**
     * If return true, skip this msg
     *
     * @param removeMap Op message map to determine whether a half message was responded by producer.
     * @param doneOpOffset Op Message which has been checked.
     * @param msgExt Half message
     * @return Return true if put success, otherwise return false.
     */
    private boolean checkPrepareQueueOffset(HashMap<Long, Long> removeMap, List<Long> doneOpOffset,
        MessageExt msgExt, String checkImmunityTimeStr) {
        String prepareQueueOffsetStr = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET);
        if (null == prepareQueueOffsetStr) {
            return putImmunityMsgBackToHalfQueue(msgExt);
        } else {
            long prepareQueueOffset = getLong(prepareQueueOffsetStr);
            if (-1 == prepareQueueOffset) {
                return false;
            } else {
                if (removeMap.containsKey(prepareQueueOffset)) {
                    long tmpOpOffset = removeMap.remove(prepareQueueOffset);
                    doneOpOffset.add(tmpOpOffset);
                    log.info("removeMap contain prepareQueueOffset. real_topic={},uniqKey={},immunityTime={},offset={}",
                            msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC),
                            msgExt.getUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX),
                            checkImmunityTimeStr,
                            msgExt.getQueueOffset());
                    return true;
                } else {
                    return putImmunityMsgBackToHalfQueue(msgExt);
                }
            }
        }
    }

    /**
     * 将messageExt再次写到HALF主题中
     *
     * @param messageExt Message will be write back to queue
     * @return Put result can used to determine the specific results of storage.
     */
    private PutMessageResult putBackToHalfQueueReturnResult(MessageExt messageExt) {
        PutMessageResult putMessageResult = null;
        try {
            // 将MessageExt转换为MessageExtBrokerInner消息
            MessageExtBrokerInner msgInner = transactionalMessageBridge.renewHalfMessageInner(messageExt);
            //
            putMessageResult = transactionalMessageBridge.putMessageReturnResult(msgInner);
        } catch (Exception e) {
            log.warn("PutBackToHalfQueueReturnResult error", e);
        }
        return putMessageResult;
    }

    private boolean putImmunityMsgBackToHalfQueue(MessageExt messageExt) {
        MessageExtBrokerInner msgInner = transactionalMessageBridge.renewImmunityHalfMessageInner(messageExt);
        return transactionalMessageBridge.putMessage(msgInner);
    }

    /**
     * 从HALF主题读取HALF消息
     *
     * @param mq Target message queue, in this method, it means the half message queue.
     * @param offset Offset in the message queue.
     * @param nums Pull message number.
     * @return Messages pulled from half message queue.
     */
    private PullResult pullHalfMsg(MessageQueue mq, long offset, int nums) {
        return transactionalMessageBridge.getHalfMessage(mq.getQueueId(), offset, nums);
    }

    /**
     * 从OP主题读取OP消息
     *
     * @param mq 目标消息队列
     * @param offset 消息队列中的偏移量
     * @param nums 拉取消息的个数
     * @return 从消息队列中拉取消息的结果
     */
    private PullResult pullOpMsg(MessageQueue mq, long offset, int nums) {
        return transactionalMessageBridge.getOpMessage(mq.getQueueId(), offset, nums);
    }

    private Long getLong(String s) {
        long v = -1;
        try {
            v = Long.parseLong(s);
        } catch (Exception e) {
            log.error("GetLong error", e);
        }
        return v;

    }

    private Integer getInt(String s) {
        int v = -1;
        try {
            v = Integer.parseInt(s);
        } catch (Exception e) {
            log.error("GetInt error", e);
        }
        return v;

    }

    private long calculateOpOffset(List<Long> doneOffset, long oldOffset) {
        Collections.sort(doneOffset);
        long newOffset = oldOffset;
        for (int i = 0; i < doneOffset.size(); i++) {
            if (doneOffset.get(i) == newOffset) {
                newOffset++;
            } else {
                break;
            }
        }
        return newOffset;

    }

    /**
     * 获取OP消息队列
     * @param messageQueue
     * @return
     */
    private MessageQueue getOpQueue(MessageQueue messageQueue) {
        // 根据消息队列获取对应的OP消息队列
        MessageQueue opQueue = opQueueMap.get(messageQueue);
        if (opQueue == null) {
            // 如果OP队列不存在，则创建OP消息队列："RMQ_SYS_TRANS_OP_HALF_TOPIC"
            // 创建Op主题名称，获取入参MQ所在的brokerName，获取入参MQ的ID
            opQueue = new MessageQueue(TransactionalMessageUtil.buildOpTopic(), messageQueue.getBrokerName(),
                messageQueue.getQueueId());
            // 将OP消息队列放到OP消息队列map集合中
            opQueueMap.put(messageQueue, opQueue);
        }
        // 返回找到的OP消息队列
        return opQueue;
    }

    /**
     * 获取HLAF消息
     * @param messageQueue
     * @param offset
     * @return
     */
    private GetResult getHalfMsg(MessageQueue messageQueue, long offset) {
        GetResult getResult = new GetResult();

        // 拉取消息
        PullResult result = pullHalfMsg(messageQueue, offset, PULL_MSG_RETRY_NUMBER);
        if (result != null) {
            getResult.setPullResult(result);
            List<MessageExt> messageExts = result.getMsgFoundList();
            if (messageExts == null || messageExts.size() == 0) {
                return getResult;
            }
            getResult.setMsg(messageExts.get(0));
        }
        return getResult;
    }

    /**
     * 通过偏移量获取HALF消息，并封装为OperationResult对象
     * @param commitLogOffset
     * @return
     */
    private OperationResult getHalfMessageByOffset(long commitLogOffset) {
        OperationResult response = new OperationResult();
        // 通过偏移量加载HALF消息
        MessageExt messageExt = this.transactionalMessageBridge.lookMessageByOffset(commitLogOffset);
        if (messageExt != null) {
            // 设置消息
            response.setPrepareMessage(messageExt);
            // 设置响应码
            response.setResponseCode(ResponseCode.SUCCESS);
        } else {
            // 如果没找到，则设置响应码：系统错误
            response.setResponseCode(ResponseCode.SYSTEM_ERROR);
            // 设置说明
            response.setResponseRemark("Find prepared transaction message failed");
        }
        // 返回响应对象
        return response;
    }

    /**
     * 将HALF消息转换为OP批消息并保存到主题中
     * @param messageExt HALF消息
     * @return
     */
    @Override
    public boolean deletePrepareMessage(MessageExt messageExt) {

        // 1. 准备和获取OP消息队列上下文：MessageQueueOpContext

        // 获取HALF消息的queueId
        Integer queueId = messageExt.getQueueId();
        // 根据HALF消息的queueId从待删除上下文中找到对应的OP消息队列上下文
        // deleteContext中：<key：half queueId> -> <value：OP消息队列上下文>
        MessageQueueOpContext mqContext = deleteContext.get(queueId);
        // 如果没有就创建一个，并放到待删除上下文中，如果失败，就使用现有的。
        if (mqContext == null) {
            // 如果没有，就创建新的
            mqContext = new MessageQueueOpContext(System.currentTimeMillis(), 20000);
            // 放到删除上下文中
            MessageQueueOpContext old = deleteContext.putIfAbsent(queueId, mqContext);
            if (old != null) {
                // 如果有返回值，表示当前添加键值对失败，返回值是现有的。
                // 这里可能是由于其他线程添加进去了
                mqContext = old;
            }
        }

        // 2. 使用HALF消息偏移量构建OP消息数据片段，并缓存到LinkedBlockingQueue中。
        // 如果有必要，则唤醒transactionalOpBatchService服务线程对OP消息进行处理

        // HALF消息的偏移量+","，OP消息体，构建需要添加到OP消息中的数据片段，
        // 一个OP消息对应多个HALF消息的偏移量信息，称OP消息为OP批消息
        // 这里的批指的是多个HALF消息偏移量构建的数据。
        String data = messageExt.getQueueOffset() + TransactionalMessageUtil.OFFSET_SEPARATOR;
        try {
            // 将data放到上下文Queue中（LinkedBlockingQueue<String>）
            boolean res = mqContext.getContextQueue().offer(data, 100, TimeUnit.MILLISECONDS);
            if (res) {
                // 记录新的数据长度
                int totalSize = mqContext.getTotalSize().addAndGet(data.length());
                // 当OP消息堆积到阈值，就唤醒事务OP批服务进行处理
                if (totalSize > transactionalMessageBridge.getBrokerController().getBrokerConfig().getTransactionOpMsgMaxSize()) {
                    this.transactionalOpBatchService.wakeup();
                }
                return true;
            } else {
                // 如果向上下文Queue中添加失败，也唤醒事务OP批服务进行处理
                this.transactionalOpBatchService.wakeup();
            }
        } catch (InterruptedException ignore) {
        }

        // 3. 将OP批消息写到主题中

        // 根据上下文数据和新数据构建OP批消息（这的批指的是HALF消息的批）
        Message msg = getOpMessage(queueId, data);
        // 如果写OP消息成功，则返回true。
        if (this.transactionalMessageBridge.writeOp(queueId, msg)) {
            log.warn("Force add remove op data. queueId={}", queueId);
            return true;
        } else {
            log.error("Transaction op message write failed. messageId is {}, queueId is {}", messageExt.getMsgId(), messageExt.getQueueId());
            return false;
        }
    }

    /**
     * 通过偏移量获取HALF消息，并封装为OperationResult对象返回。
     * @param requestHeader Commit message request header.
     * @return 返回操作的结果以及对应的错误码
     */
    @Override
    public OperationResult commitMessage(EndTransactionRequestHeader requestHeader) {
        // 通过偏移量获取HALF消息
        return getHalfMessageByOffset(requestHeader.getCommitLogOffset());
    }

    /**
     * 根据偏移量获取HALF消息并封装为OperationResult对象返回。
     * @param requestHeader Prepare message request header.
     * @return
     */
    @Override
    public OperationResult rollbackMessage(EndTransactionRequestHeader requestHeader) {
        // 根据偏移量获取HALF消息并封装为OperationResult对象返回。
        return getHalfMessageByOffset(requestHeader.getCommitLogOffset());
    }

    @Override
    public boolean open() {
        return true;
    }

    @Override
    public void close() {
        if (this.transactionalOpBatchService != null) {
            this.transactionalOpBatchService.shutdown();
        }
    }

    /**
     * 根据上下文数据和新数据构建OP批消息（这的批指的是HALF消息的批）
     * @param queueId
     * @param moreData
     * @return
     */
    public Message getOpMessage(int queueId, String moreData) {
        // 获取OP主题名称："RMQ_SYS_TRANS_OP_HALF_TOPIC"
        String opTopic = TransactionalMessageUtil.buildOpTopic();
        // 根据queueId获取OP消息队列上下文
        MessageQueueOpContext mqContext = deleteContext.get(queueId);

        // 如果构建的OP数据内容不是null，则返回真正的长度，否则返回0。
        int moreDataLength = moreData != null ? moreData.length() : 0;
        // 赋值给长度
        int length = moreDataLength;
        // 获取OP消息最大长度（默认4K）
        int maxSize = transactionalMessageBridge.getBrokerController().getBrokerConfig().getTransactionOpMsgMaxSize();

        // 如果数据长度小于最大长度
        if (length < maxSize) {
            // 获取上下文队列中的数据总长度
            int sz = mqContext.getTotalSize().get();

            // 如果数据总长度大于最大消息长度或者当前数据长度+数据总长度大于最大长度，则在最大长度的基础上在加上100
            if (sz > maxSize || length + sz > maxSize) {
                length = maxSize + 100;
            } else {
                // 否则长度加上上下文数据总长度
                length += sz;
            }
        }

        // 构建字符串
        StringBuilder sb = new StringBuilder(length);
        // 如果有新数据，则追加新数据
        if (moreData != null) {
            sb.append(moreData);
        }

        // 将上下文队列中的数据追加到stringbuffer
        while (!mqContext.getContextQueue().isEmpty()) {
            // 如果大于等于最大长度，则跳出循环
            if (sb.length() >= maxSize) {
                break;
            }
            String data = mqContext.getContextQueue().poll();
            if (data != null) {
                sb.append(data);
            }
        }

        // 如果拼接的stringbuffer长度是0，则返回null
        if (sb.length() == 0) {
            return null;
        }

        // stringbuffer长度减去新数据长度
        int l = sb.length() - moreDataLength;

        // 将上下文队列的总数据长度减去该差值
        mqContext.getTotalSize().addAndGet(-l);
        // 设置上下文的最后写入时间
        mqContext.setLastWriteTimestamp(System.currentTimeMillis());
        // 返回OP批消息（这里批的意思是HALF消息数据的批）
        return new Message(opTopic, TransactionalMessageUtil.REMOVE_TAG,
                sb.toString().getBytes(TransactionalMessageUtil.CHARSET));
    }

    /**
     * 发送OP消息
     * @return
     */
    public long batchSendOpMessage() {
        long startTime = System.currentTimeMillis();
        try {
            long firstTimestamp = startTime;
            Map<Integer, Message> sendMap = null;

            // 默认：3000
            long interval = transactionalMessageBridge.getBrokerController().getBrokerConfig().getTransactionOpBatchInterval();
            // 默认：4096
            int maxSize = transactionalMessageBridge.getBrokerController().getBrokerConfig().getTransactionOpMsgMaxSize();
            // OP消息是否超过最大长度阈值
            boolean overSize = false;

            // 遍历
            for (Map.Entry<Integer, MessageQueueOpContext> entry : deleteContext.entrySet()) {
                MessageQueueOpContext mqContext = entry.getValue();
                //no msg in contextQueue
                if (mqContext.getTotalSize().get() <= 0 || mqContext.getContextQueue().size() == 0 ||
                        // wait for the interval
                        mqContext.getTotalSize().get() < maxSize &&
                                startTime - mqContext.getLastWriteTimestamp() < interval) {
                    continue;
                }

                if (sendMap == null) {
                    sendMap = new HashMap<>();
                }

                Message opMsg = getOpMessage(entry.getKey(), null);
                if (opMsg == null) {
                    continue;
                }
                sendMap.put(entry.getKey(), opMsg);
                firstTimestamp = Math.min(firstTimestamp, mqContext.getLastWriteTimestamp());
                if (mqContext.getTotalSize().get() >= maxSize) {
                    overSize = true;
                }
            }

            if (sendMap != null) {
                for (Map.Entry<Integer, Message> entry : sendMap.entrySet()) {
                    if (!this.transactionalMessageBridge.writeOp(entry.getKey(), entry.getValue())) {
                        log.error("Transaction batch op message write failed. body is {}, queueId is {}",
                                new String(entry.getValue().getBody(), TransactionalMessageUtil.CHARSET), entry.getKey());
                    }
                }
            }

            log.debug("Send op message queueIds={}", sendMap == null ? null : sendMap.keySet());

            //wait for next batch remove
            long wakeupTimestamp = firstTimestamp + interval;
            if (!overSize && wakeupTimestamp > startTime) {
                return wakeupTimestamp;
            }
        } catch (Throwable t) {
            log.error("batchSendOp error.", t);
        }

        return 0L;
    }

    public Map<Integer, MessageQueueOpContext> getDeleteContext() {
        return this.deleteContext;
    }
}
